#include "Net.h"
#include <string.h>
#include <atomic>

#if FG_PLATFORM == FG_PLATFORM_WIN
#include <WS2tcpip.h>
#include <winsock2.h>
#pragma  comment(lib,"Ws2_32.lib")
#ifndef FG_LIBEVENT_SRC
#pragma  comment(lib,"event.lib")
#pragma  comment(lib,"event_core.lib")
#pragma  comment(lib,"event_extra.lib")
#pragma  comment(lib,"event_openssl.lib")
#endif
#elif FG_PLATFORM == FG_PLATFORM_APPLE
#include <arpa/inet.h>
#endif

#include <iostream>

#include <event2/event.h>
#include <event2/bufferevent_struct.h>

//1048576 = 1024 * 1024
#define GF_BUFFER_MAX_READ	1048576

FragrantGrass::Net::Net()
{
	m_run_as_server = false;
	m_tcpStream = false;
	m_get_ready = false;
	m_port = 0;
	m_ip.clear();
	m_event_base = NULL;
	m_cb_read = NULL;
	m_cb_event = NULL;
	m_msg_recieved_cnt = 0;
	m_max_client_cnt = 0;
	m_cpu_count = 0;
	m_listener = NULL;
	m_buffer_size = 0;
}

FragrantGrass::Net::~Net()
{

}

int FragrantGrass::Net::init_as_client(const char* ip, unsigned short port)
{
	m_ip = ip;
	m_port = port;

	return init_client_net(m_ip, m_port);
}

int FragrantGrass::Net::init_as_server(unsigned int max_client_cnt, unsigned short port, int cpu_count)
{
	m_max_client_cnt = max_client_cnt;
	m_port = port;
	m_cpu_count = cpu_count;

	return init_server_net();
}

int FragrantGrass::Net::expand_buffer_size(unsigned int size)
{
	if (size > 0)
	{
		m_buffer_size = size;
	}
	return m_buffer_size;
}

bool FragrantGrass::Net::execute()
{
	execute_close();

	if (m_event_base)
	{
		event_base_loop(m_event_base, EVLOOP_ONCE | EVLOOP_NONBLOCK);
	}

	return true;
}

bool FragrantGrass::Net::destroy()
{
	close_all_sockets();

	if (m_listener)
	{
		evconnlistener_free(m_listener);
		m_listener = NULL;
	}

	if (m_event_base)
	{
		event_base_free(m_event_base);
		m_event_base = NULL;
	}

	return true;
}

bool FragrantGrass::Net::run_as_server()
{
	return m_run_as_server;
}

void FragrantGrass::Net::listener_cb(struct evconnlistener* listener, evutil_socket_t fd, struct sockaddr* sa, int socklen, void* user_data)
{
	Net* pNet = (Net*)user_data;
	bool closed = pNet->close_net_handle(fd);
	if (closed)
	{
		return;
	}

	if (pNet->m_net_handle_impls.size() >= pNet->m_max_client_cnt)
	{
		return;
	}

	struct event_base* mx_event_base = pNet->m_event_base;

	struct bufferevent* bev = bufferevent_socket_new(mx_event_base, fd, BEV_OPT_CLOSE_ON_FREE);
	if (!bev)
	{
		fprintf(stderr, "Error constructing bufferevent!");
		return;
	}

	struct sockaddr_in* pSin = (sockaddr_in*)sa;

	NetHandle* pHandle = new NetHandle(pNet, fd, *pSin, bev);
	pHandle->getNet()->add_net_handle(fd, pHandle);

	bufferevent_setcb(bev, conn_readcb, conn_writecb, conn_eventcb, (void*)pHandle);

	bufferevent_enable(bev, EV_READ | EV_WRITE | EV_CLOSED | EV_TIMEOUT | EV_PERSIST);

	event_set_fatal_callback(event_fatal_cb);

	conn_eventcb(bev, BEV_EVENT_CONNECTED, (void*)pHandle);

	bufferevent_set_max_single_read(bev, GF_BUFFER_MAX_READ);
	bufferevent_set_max_single_write(bev, GF_BUFFER_MAX_READ);
}

void FragrantGrass::Net::conn_readcb(struct bufferevent* bev, void* user_data)
{
	NetHandle* pHandle = (NetHandle*)user_data;
	if (!pHandle)
	{
		return;
	}

	Net* pNet = (Net*)pHandle->getNet();
	if (!pNet)
	{
		return;
	}

	if (pNet->get_need_removed())
	{
		return;
	}

	struct evbuffer* input = bufferevent_get_input(bev);
	if (!input)
	{
		return;
	}

	size_t len = evbuffer_get_length(input);
	unsigned char* pData = evbuffer_pullup(input, len);
	pHandle->append((const char*)pData, len);
	evbuffer_drain(input, len);

	if (pNet->m_tcpStream)
	{
		int len = pHandle->length();
		if (len > 0)
		{
			if (pNet->m_cb_read)
			{
				pNet->m_cb_read(pHandle->fd(), 0, pHandle->data(), len);

				++pNet->m_msg_recieved_cnt;
			}
			pHandle->remove(0, len);
		}
	}
	else
	{
		while (1)
		{
			if (!pNet->dismantle(pHandle))
			{
				break;
			}
		}
	}
}

void FragrantGrass::Net::conn_writecb(struct bufferevent* bev, void* user_data)
{

}

void FragrantGrass::Net::conn_eventcb(struct bufferevent* bev, short events, void* user_data)
{
	NetHandle* pHandle = (NetHandle*)user_data;
	Net* pNet = (Net*)pHandle->getNet();
	
	std::thread::id id = std::this_thread::get_id();

	std::cout << "[conn_eventcb]:" << " thread_id:" << id << " fd:" << pHandle->fd() << " event_id:" << events << std::endl;

	if (events & BEV_EVENT_CONNECTED)
	{
		//must to set it's state before the "EventCB" functional be called[maybe user will send msg in the callback function]
		pNet->m_get_ready = true;
		printf("[conn_eventcb]: connect operation finished eventid = %d\n", BEV_EVENT_CONNECTED);
	}
	else
	{
		if (events & BEV_EVENT_READING)
		{
			fprintf(stderr, "[conn_eventcb]: error encountered while reading eventid = %d\n", BEV_EVENT_READING);
		}
		if (events & BEV_EVENT_WRITING)
		{
			fprintf(stderr, "[conn_eventcb]: error encountered while writing eventid = %d\n", BEV_EVENT_WRITING);
		}
		if (events & BEV_EVENT_EOF)
		{
			fprintf(stderr, "[conn_eventcb]: eof file reached eventid = %d\n", BEV_EVENT_EOF);
		}
		if (events & BEV_EVENT_ERROR)
		{
			fprintf(stderr, "[conn_eventcb]: unrecoverable error encountered eventid = %d\n", BEV_EVENT_ERROR);
		}
		if (events & BEV_EVENT_TIMEOUT)
		{
			fprintf(stderr, "[conn_eventcb]: user-specified timeout reached eventid = %d\n", BEV_EVENT_TIMEOUT);
		}

		if (!pNet->m_run_as_server)
		{
			pNet->m_get_ready = false;
		}
	}

	if (pNet->m_cb_event)
	{
		pNet->m_cb_event(pHandle->fd(), FG_NET_EVENT(events), pNet);
	}

	if (events & BEV_EVENT_CONNECTED)
	{
		struct evbuffer* input = bufferevent_get_input(bev);
		struct evbuffer* output = bufferevent_get_output(bev);
		if (pNet->m_buffer_size > 0)
		{
			evbuffer_expand(input, pNet->m_buffer_size);
			evbuffer_expand(output, pNet->m_buffer_size);
		}
	}
	else
	{
		pNet->close_net_handle(pHandle->fd());
	}
}

void FragrantGrass::Net::log_cb(int severity, const char* msg)
{
	fprintf(stdout, "severity:[%d] msg:%s\n", severity, msg);
}

void FragrantGrass::Net::event_fatal_cb(int err)
{

}

bool FragrantGrass::Net::add_net_handle(SOCK fd, NetHandle* handle)
{
	if (!handle)
		return false;
	return m_net_handle_impls.insert(NetHandleMap::value_type(fd, handle)).second;;
}

bool FragrantGrass::Net::close_net_handle(SOCK fd)
{
	NetHandleMap::iterator it = m_net_handle_impls.find(fd);
	if (it == m_net_handle_impls.end())
	{
		return false;
	}

	NetHandle* pHandle = it->second;
	pHandle->set_need_removed(true);
	m_removed_net_handles.push_back(fd);

	return true;
}

bool FragrantGrass::Net::send_msg(const char* msg, uint32_t len, SOCK fd)
{
	NetHandleMap::iterator it = m_net_handle_impls.find(fd);
	if (it == m_net_handle_impls.end())
	{
		return false;
	}

	NetHandle* pHandle = it->second;
	if (pHandle->get_need_removed())
	{
		return false;
	}

	struct bufferevent* bev = (struct bufferevent*)pHandle->get_user_data();
	if (NULL == bev)
	{
		return false;
	}

	bufferevent_write(bev, msg, len);

	return true;
}

bool FragrantGrass::Net::send_msg_with_out_head(int16_t id, const char* msg, uint32_t len, SOCK fd)
{
	NetHandleMap::iterator it = m_net_handle_impls.find(fd);
	if (it == m_net_handle_impls.end())
	{
		return false;
	}

	int verify_length = (int)(len + MsgHead::Head::HEAD_LENGTH);
	std::string out_data;
	int encode_length = encode(id, msg, len, out_data);
	if (encode_length != verify_length)
	{
		return false;
	}

	return send_msg(out_data.data(),(uint32_t)out_data.length(), fd);
}

bool FragrantGrass::Net::send_msg_to_all_client(const char* msg, uint32_t len)
{
	if(!m_get_ready)
	{
		return false;
	}

	NetHandleMap::iterator it = m_net_handle_impls.begin();
	NetHandleMap::const_iterator end = m_net_handle_impls.end();
	while (it != end)
	{
		NetHandle* pHandle = it->second;
		if (pHandle->get_need_removed())
		{
			continue;
		}

		struct bufferevent* bev = (struct bufferevent*)pHandle->get_user_data();
		if (NULL == bev)
		{
			continue;
		}

		bufferevent_write(bev, msg, len);

		++it;
	}

	return true;
}

bool FragrantGrass::Net::send_msg_to_all_client_with_out_head(int16_t id, const char* msg, uint32_t len)
{
	int verify_length = (int)(len + MsgHead::Head::HEAD_LENGTH);
	std::string out_data;
	int encode_length = encode(id, msg, len, out_data);
	if (encode_length != verify_length)
	{
		return false;
	}

	return send_msg_to_all_client(out_data.data(),(uint32_t)out_data.length());
}

bool FragrantGrass::Net::send_msg_to_specific_clients(const char* msg, uint32_t len, const std::list<SOCK>& clients)
{
	std::list<SOCK>::const_iterator end = clients.end();
	for (std::list<SOCK>::const_iterator it = clients.begin(); it != end; ++it)
	{
		send_msg(msg, len, *it);
	}
	return true;
}

bool FragrantGrass::Net::send_msg_to_specific_clients_with_out_head(int16_t id, const char* msg, uint32_t len, const std::list<SOCK>& clients)
{
	int verify_length = (int)(len + MsgHead::Head::HEAD_LENGTH);
	std::string out_data;
	int encode_length = encode(id, msg, len, out_data);
	if (encode_length != verify_length)
	{
		return false;
	}

	return send_msg_to_specific_clients(msg,len, clients);
}

bool FragrantGrass::Net::dismantle(NetHandle* pHandle)
{
	assert(NULL != pHandle);

	int len = pHandle->length();
	if (len <= IMsgHead::Head::HEAD_LENGTH)
	{
		//msg length not length enough
		return false;
	}

	MsgHead head;
	int msg_body_length = decode(pHandle->data(), len, head);
	if (msg_body_length <= 0)
	{
		return false;
	}

	if (m_cb_read)
	{
		m_cb_read(pHandle->fd(), head.get_msg_id(), pHandle->data() + IMsgHead::Head::HEAD_LENGTH, msg_body_length);
	}

	++m_msg_recieved_cnt;
	pHandle->remove(0, msg_body_length + IMsgHead::Head::HEAD_LENGTH);

	return true;
}

int FragrantGrass::Net::decode(const char* data, uint32_t len, MsgHead& head)
{
	if (len < IMsgHead::Head::HEAD_LENGTH)
	{
		return -1;
	}

	if (IMsgHead::Head::HEAD_LENGTH != head.decode(data))
	{
		return -2;
	}

	if (head.get_body_length() > (len - IMsgHead::Head::HEAD_LENGTH))
	{
		return -3;
	}

	return head.get_body_length();
}

int FragrantGrass::Net::encode(uint16_t id, const char* data, uint32_t len, std::string& out_data)
{
	MsgHead msg_head;
	msg_head.set_msg_id(id);
	msg_head.set_body_length(len);

	char szHead[MsgHead::Head::HEAD_LENGTH] = { 0 };
	msg_head.encode(szHead);

	out_data.clear();
	out_data.append(szHead, MsgHead::Head::HEAD_LENGTH);
	out_data.append(data, len);

	return msg_head.get_body_length() + MsgHead::Head::HEAD_LENGTH;
}

int FragrantGrass::Net::init_client_net(const std::string& ip, int port)
{
	struct sockaddr_in addr;
	struct bufferevent* bev = NULL;

#ifdef _MSC_VER
	WSADATA wsa_data;
	int errCode = WSAStartup(0x0201, &wsa_data);
	if (errCode != 0)
	{
		printf("WSAStartup Error = %d\n", errCode);
		return -1;
	}
#endif

	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);

	//convert ip string from host byteorder to net byteorder
	if (evutil_inet_pton(AF_INET, ip.c_str(), &addr.sin_addr) <= 0)
	{
		printf("inet_pton_error\n");
		return -1;
	}

	m_event_base = event_base_new();
	if (0 == m_event_base)
	{
		printf("event_base_new failed\n");
		return -1;
	}

	bev = bufferevent_socket_new(m_event_base, -1, BEV_OPT_CLOSE_ON_FREE);
	if (bev == NULL)
	{
		printf("bufferevent_socket_new\n");
		return -1;
	}

	int bRet = bufferevent_socket_connect(bev, (struct sockaddr*) & addr, sizeof(addr));
	if (0 != bRet)
	{
		printf("bufferevent_socket_connect error\n");
		return -1;
	}

	SOCK sockfd = bufferevent_getfd(bev);
	NetHandle* pHandle = new NetHandle(this, sockfd, addr, bev);

	if (!add_net_handle(0, pHandle))
	{
		assert(0);
		return -1;
	}

	m_run_as_server = false;

	bufferevent_setcb(bev, conn_readcb, conn_writecb, conn_eventcb, (void*)pHandle);
	bufferevent_enable(bev, EV_READ | EV_WRITE | EV_CLOSED | EV_TIMEOUT | EV_PERSIST);
	event_set_log_callback(&Net::log_cb);

	bufferevent_set_max_single_read(bev, GF_BUFFER_MAX_READ);
	bufferevent_set_max_single_write(bev, GF_BUFFER_MAX_READ);

	int size_to_read = (int)bufferevent_get_max_to_read(bev);
	int size_to_write = (int)bufferevent_get_max_to_write(bev);

	fprintf(stdout, "[init_client_net]:ready to connect %s\n[init_client_net]:read_io_size:%d write_io_size:%d\n", m_ip.c_str(), size_to_read, size_to_write);

	return (int)sockfd;
}

int FragrantGrass::Net::init_server_net()
{
	int nCpuCount = m_cpu_count;
	int nPort = m_port;

	struct sockaddr_in sin;

#if FG_PLATFORM == FG_PLATFORM_WIN
	WSADATA wsa_data;
	int errCode = WSAStartup(0x0201, &wsa_data);
	if (errCode != 0)
	{
		fprintf(stderr,"WSAStartup Error = %d\n", errCode);
		destroy();
		return -1;
	}
#endif
	//////////////////////////////////////////////////////////////////////////

	struct event_config* cfg = event_config_new();

#if FG_PLATFORM == FG_PLATFORM_WIN
	m_event_base = event_base_new_with_config(cfg);
#else

	//event_config_avoid_method(cfg, "epoll");
	if (event_config_set_flag(cfg, EVENT_BASE_FLAG_EPOLL_USE_CHANGELIST) < 0)
	{

		return -1;
	}

	if (event_config_set_num_cpus_hint(cfg, nCpuCount) < 0)
	{
		return -1;
	}

	m_event_base = event_base_new_with_config(cfg);//event_base_new()

#endif
	event_config_free(cfg);

	//////////////////////////////////////////////////////////////////////////

	if (!m_event_base)
	{
		fprintf(stderr, "Could not initialize libevent!\n");
		destroy();

		return -1;
	}

	memset(&sin, 0, sizeof(sin));
	sin.sin_family = AF_INET;
	sin.sin_port = htons(nPort);

	printf("server started with %d\n", nPort);

	m_listener = evconnlistener_new_bind(m_event_base, listener_cb, (void*)this,
		LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE, -1,
		(struct sockaddr*) & sin,
		sizeof(sin));

	if (!m_listener)
	{
		fprintf(stderr, "Could not create a listener!\n");
		destroy();

		return -1;
	}

	m_run_as_server = true;

	event_set_log_callback(&Net::log_cb);

	return m_max_client_cnt;
}

bool FragrantGrass::Net::close_all_sockets()
{
	NetHandleMap::iterator it = m_net_handle_impls.begin();
	for (; it != m_net_handle_impls.end(); ++it)
	{
		SOCK fd = it->first;
		m_removed_net_handles.push_back(fd);
	}

	execute_close();

	m_net_handle_impls.clear();

	return true;
}

void FragrantGrass::Net::execute_close()
{
	int n = (int)m_removed_net_handles.size();
	for (int i = 0; i < n; ++i)
	{
		SOCK fd = m_removed_net_handles[i];
		close_net_handle_immediately(fd);
	}
	m_removed_net_handles.clear();
}

void FragrantGrass::Net::close_net_handle_immediately(SOCK fd)
{
	NetHandleMap::iterator it = m_net_handle_impls.find(fd);
	if (it != m_net_handle_impls.end())
	{
		NetHandle* pHandle = it->second;
		struct bufferevent* bev = (bufferevent*)pHandle->get_user_data();
		bufferevent_free(bev);
		m_net_handle_impls.erase(it);
		delete pHandle;
		pHandle = NULL;
	}
}